package job.relevance.rawpage;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import mapred.base.TextMapper;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

public class RLV_RawPage_Mapper extends TextMapper {

    private static Map<String, List<String>> map;

    @Override
    public void map(Text key, Text value) throws IOException {
        List<String> list = map.get(key.toString().toLowerCase());
        if (list != null) {
            for (String s : list) {
                this.output(new Text(s), value);
            }
        }
    }

    @Override
    public void setUp() {
        map = new HashMap<String, List<String>>();

        try {
            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(this.getConfig("list")))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                String k = line.split("\t")[2].toLowerCase();
                if (!map.containsKey(k)) {
                    map.put(k, new ArrayList<String>());
                }
                map.get(k).add(line);
            }
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(255);
        }
    }

    @Override
    public void tearDown() throws IOException {
    }

}
